package com.example.study1206.controller;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.ConsumerGroupState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

@RestController
public class HelloController {

    @Autowired
    private AdminClient adminClient;


    @GetMapping("/searchTopic/{message}")
    public String searchTopic(@PathVariable("message") String topicName) throws InterruptedException {
        DescribeTopicsResult result = adminClient.describeTopics(
                Arrays.asList(topicName));
        String shownResult;
        try {
            shownResult = result.all().get().values().toString();
        } catch (ExecutionException e) {
            return "该topic不存在";
        }

        return shownResult;
//        adminClient.createTopics(Arrays.asList(new NewTopic("yeah",4,(short) 1)));

//        DeleteTopicsResult result0 = adminClient.deleteTopics(Arrays.asList("yeah"));
//        result0.topicNameValues().forEach((k,v)->System.out.println("k: "+k+" ,v: "+v.toString()+"\n"));

//        ListTopicsResult result =  adminClient.listTopics();
//        result.namesToListings().get().forEach((k,v)->System.out.println("k: "+k+" ,v: "+v.toString()+"\n"));
    }

    @GetMapping("/topicList")
    public Collection<String> getTopicList() throws ExecutionException, InterruptedException {
        ListTopicsResult result = adminClient.listTopics();
        return result.names().get();

    }

    @PostMapping("/addNewPartition/{topicName}/{number}")
    public String addPartition(@PathVariable("topicName") String topicName, @PathVariable("number") int number) {
        Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
        newPartitionsMap.put(topicName, NewPartitions.increaseTo(number));
        CreatePartitionsResult result = adminClient.createPartitions(newPartitionsMap);
        result.values().forEach((k, v) -> System.out.println("k: " + k + " ,v: " + v.toString() + "\n"));
        return "添加分区成功";
    }

    @GetMapping("/groupList")
    public Collection<String> getGroupList() throws ExecutionException, InterruptedException {
        ListConsumerGroupsResult result = adminClient.listConsumerGroups();
        Collection<String> shownResult = result.all().get().stream().map(p -> p.groupId()).collect(Collectors.toList());
        return shownResult;
    }

    @GetMapping("/searchGroup/{message}")
    public String searchGroup(@PathVariable("message") String GroupName) throws ExecutionException, InterruptedException {
//        ListConsumerGroupsResult listResult = adminClient.listConsumerGroups();
//        Collection<String> GroupList = listResult.all().get().stream().map(p->p.groupId()).collect(Collectors.toList());
//        if(GroupList.contains(GroupName)){
//        DescribeConsumerGroupsResult result = adminClient.describeConsumerGroups(Arrays.asList(GroupName));
//        String shownResult = result.all().get().values().toString();
//        return shownResult;}else {
//            return "该group不存在";
//        }
        DescribeConsumerGroupsResult result = adminClient.describeConsumerGroups(Arrays.asList(GroupName));
        if (result.all().get().get(GroupName).state() == ConsumerGroupState.DEAD) {
            return "该group不存在";
        } else {
            String shownResult = result.all().get().values().toString();
            return shownResult;
        }
    }


    @GetMapping("equipAllTopic")
    public void equip() {
        adminClient.createTopics(Arrays.asList(new NewTopic("Hydrogen", 4, (short) 1),
                new NewTopic("Helium", 4, (short) 1),
                new NewTopic("Lithium", 4, (short) 1),
                new NewTopic("Beryllium", 4, (short) 1),
                new NewTopic("Boron", 4, (short) 1),
                new NewTopic("Carbon", 4, (short) 1),
                new NewTopic("Nitrogen", 4, (short) 1),
                new NewTopic("Oxygen", 4, (short) 1),
                new NewTopic("Fluorine", 4, (short) 1),
                new NewTopic("Neon", 4, (short) 1)));
    }

}
